package com.lee.trigger;

/**
 * Created by liruopeng on 2018/1/18.
 */

import com.lee.IP;
import com.lee.TCP;
import com.lee.UDP;
import com.lee.entity.SessionResult;
import com.lee.networkData;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.Window;

import java.util.HashMap;

/**
 * A {@link Trigger} that fires once the count of elements in a pane reaches the given count.
 *
 * @param <W> The type of {@link Window Windows} on which this trigger can operate.
 */
@PublicEvolving
public class SessionCountWindowTrigger<W extends Window> extends Trigger<SessionResult, W> {


    private static final long serialVersionUID = 1L;

    private final long maxCount;

    HashMap<String, Integer> countDstMap;
    HashMap<String, Integer> countServiceMap;
    long count = 0;


    public SessionCountWindowTrigger(long maxCount, HashMap<String, Integer> countDstMap, HashMap<String, Integer> countServiceMap) {
        this.maxCount = maxCount;
        this.countDstMap = countDstMap;
        this.countServiceMap = countServiceMap;
    }

    @Override
    public TriggerResult onElement(SessionResult element, long timestamp, W window, TriggerContext ctx) throws Exception {
        count+=1;
        if (count >= maxCount) {
            count = 0;
            return TriggerResult.FIRE_AND_PURGE;
        }

        otherAddOne(element);
        return TriggerResult.CONTINUE;
    }



    public void otherAddOne(SessionResult element) {
        if (countDstMap.containsKey(element.getDstIP().toString())) {
            countDstMap.put(element.getDstIP().toString(), countDstMap.get(element.getDstIP().toString()) + 1);
        } else {
            countDstMap.put(element.getDstIP().toString(), 1);
        }

        String protocol = element.getProtocol();
        if (countServiceMap.containsKey(protocol)) {
            countServiceMap.put(protocol, countServiceMap.get(protocol) + 1);
        } else {
            countServiceMap.put(protocol, 1);
        }
    }

    @Override
    public TriggerResult onEventTime(long time, W window, TriggerContext ctx) {
        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception {
        count = 0;
        return TriggerResult.FIRE_AND_PURGE;
    }

    @Override
    public void clear(W window, TriggerContext ctx) throws Exception {

    }

    @Override
    public boolean canMerge() {
        return true;
    }

    @Override
    public void onMerge(W window, OnMergeContext ctx) throws Exception {

    }

    @Override
    public String toString() {
        return "CountTrigger(" + maxCount + ")";
    }


    private static class Sum implements ReduceFunction<Long> {
        private static final long serialVersionUID = 1L;

        @Override
        public Long reduce(Long value1, Long value2) throws Exception {
            return value1 + value2;
        }

    }
}
